package com.example.flinkcourse.lesson4.state;

import com.example.flinkcourse.lesson4.model.Event;
import com.example.flinkcourse.lesson4.model.Rule;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 广播状态处理器
 * 用于处理广播状态，实现规则匹配功能
 */
public class BroadcastStateProcessor extends BroadcastProcessFunction<Event, Rule, Event> {
    private static final Logger LOG = LoggerFactory.getLogger(BroadcastStateProcessor.class);
    
    private final MapStateDescriptor<String, Rule> ruleStateDescriptor;
    
    /**
     * 构造函数
     * @param ruleStateDescriptor 规则状态描述符
     */
    public BroadcastStateProcessor(MapStateDescriptor<String, Rule> ruleStateDescriptor) {
        this.ruleStateDescriptor = ruleStateDescriptor;
    }
    
    @Override
    public void processElement(Event event, ReadOnlyContext ctx, Collector<Event> out) {
        LOG.debug("Processing event: {}", event.getId());
        
        try {
            // 获取广播状态
            ReadOnlyBroadcastState<String, Rule> ruleState = ctx.getBroadcastState(ruleStateDescriptor);
            
            // 获取事件类型对应的规则
            Rule rule = ruleState.get(event.getType());
            
            if (rule != null && rule.getEnabled()) {
                // 检查规则条件
                if (event.getAmount() > rule.getThreshold()) {
                    // 更新事件
                    event.setRule(rule);
                    event.getProperties().put("rule-matched", "true");
                    event.getProperties().put("rule-description", rule.getDescription());
                    
                    // 输出事件
                    out.collect(event);
                    
                    LOG.info("Event matched rule: {}", event.getId());
                } else {
                    LOG.debug("Event did not match rule threshold: {}", event.getId());
                }
            } else {
                LOG.debug("No rule found for event type: {}", event.getType());
            }
        } catch (Exception e) {
            LOG.error("Error processing event: " + event.getId(), e);
        }
    }
    
    @Override
    public void processBroadcastElement(Rule rule, Context ctx, Collector<Event> out) {
        LOG.info("Processing broadcast rule: {}", rule.getType());
        
        try {
            // 获取广播状态
            BroadcastState<String, Rule> ruleState = ctx.getBroadcastState(ruleStateDescriptor);
            
            // 更新规则
            ruleState.put(rule.getType(), rule);
            
            LOG.info("Rule updated successfully: {}", rule.getType());
        } catch (Exception e) {
            LOG.error("Error processing broadcast rule: " + rule.getType(), e);
        }
    }
    
    /**
     * 创建规则状态描述符
     * @return 规则状态描述符
     */
    public static MapStateDescriptor<String, Rule> createRuleStateDescriptor() {
        return new MapStateDescriptor<>(
            "RulesBroadcastState",
            String.class,
            Rule.class
        );
    }
} 